package com.hery.functions;

import com.hery.pojo.KafkaAlertLog;
import com.hery.pojo.Rule;
import com.hery.pojo.alertManager.Route;
import com.hery.services.RulesService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.SQLException;
import java.util.List;


/**
 * 规则匹配核心类
 */
@Slf4j
@Data
public class RuleMatchBroadCastProcessFunction extends CoProcessFunction<String, Rule, KafkaAlertLog> {

    private RulesService rulesService;

    private Route route;

    private static final String alertName = "Sentry Alert";

    public RuleMatchBroadCastProcessFunction(Configuration config) throws  SQLException {

        ConfigOption<Route> routeOption = ConfigOptions.key("am.route").enumType(Route.class).defaultValue(Route.WEBHOOK);

        route = config.get(routeOption);
        rulesService = new RulesService(config);
        // rulesService.allSyncRule();
        log.info("init load Mysql rule completed, total count: {} ", rulesService.getRules().size());
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void processElement1(String value, Context ctx, Collector<KafkaAlertLog> out) {
        log.debug("accept a per dada: {}", value);
        List<Rule> rules = rulesService.mathRules(value);
        for (Rule rule : rules) {
            out.collect(new KafkaAlertLog(value, rule));
        }
    }


    @Override
    public void processElement2(Rule value, Context ctx, Collector<KafkaAlertLog> out) {
        log.debug("rule refresh ：{}", value);
        rulesService.incrementSyncRule(value);
    }

}
